package org.yuanzheng.function

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}
import java.text.SimpleDateFormat
import java.util.Date

import org.apache.flink.api.common.functions.{MapFunction, RichMapFunction}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.yuanzheng.source.StationLog

/**
 * @author yuanzheng
 * @date 2020/6/17-21:01
 */
object TestRichFunctionClass {
  def main(args: Array[String]): Unit = {
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setParallelism(1)
    import org.apache.flink.streaming.api.scala._
    val path = getClass.getResource("/station.log").getPath
    //读取数据源
    val stream: DataStream[StationLog] = streamEnv.readTextFile(path).map(line => {
      var split = line.split(",")
      new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
    })
    //定义一个时间格式
    val simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss")

    //计算数据
    val result: DataStream[String] = stream.filter(_.callType.equals("success")).map(new UserDefinedRichMapFunction(simpleDateFormat))
    result.print()
    streamEnv.execute()
  }

  //自定义函数类
  class UserDefinedRichMapFunction(simpleDateFormat: SimpleDateFormat) extends RichMapFunction[StationLog, String] {
    var conn: Connection = _
    var pst: PreparedStatement = _

    override def open(parameters: Configuration): Unit = {
      conn = DriverManager.getConnection("jdbc:mysql://localhost/yuanzheng?characterEncoding=UTF-8&useUnicode=true&useSSL=false", "root", "root")
      pst = conn.prepareStatement("select name from station where phone = ?")
    }

    override def close(): Unit = {
      pst.close()
      conn.close()
    }

    override def map(t: StationLog): String = {
      //查询被叫号码
      pst.setString(1, t.callIn)
      val resultSet1: ResultSet = pst.executeQuery()
      if (resultSet1.next()) {
        t.callIn = resultSet1.getString(1)
      }
      //查询主叫号码
      pst.setString(1, t.callOut)
      val resultSet2: ResultSet = pst.executeQuery()
      if (resultSet2.next()) {
        t.callOut = resultSet2.getString(1)
      }
      var callTime = t.callTime
      var endTime = callTime + t.duration * 1000
      "主叫号码：" + t.callOut + ",被叫号码：" + t.callIn + ",呼叫开始时间：" + simpleDateFormat.format(new Date(callTime)) + ",呼叫结束时间：" + simpleDateFormat.format(new Date(endTime))
    }
  }

}
